package jp.co.worksap.ate.gomoku.hadoopai;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import jp.co.worksap.ate.gomoku.localai.GoNode;
import jp.co.worksap.ate.gomoku.localai.GobanNode;
import jp.co.worksap.ate.gomoku.localai.LocalAIUtil;
import jp.co.worksap.ate.gomoku.model.GoState;
import jp.co.worksap.ate.gomoku.model.Goban;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem.Statistics;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapred.JobConf;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.ec2.model.InstanceType;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduce;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient;
import com.amazonaws.services.elasticmapreduce.model.DescribeJobFlowsRequest;
import com.amazonaws.services.elasticmapreduce.model.DescribeJobFlowsResult;
import com.amazonaws.services.elasticmapreduce.model.HadoopJarStepConfig;
import com.amazonaws.services.elasticmapreduce.model.JobFlowDetail;
import com.amazonaws.services.elasticmapreduce.model.JobFlowExecutionState;
import com.amazonaws.services.elasticmapreduce.model.JobFlowInstancesConfig;
import com.amazonaws.services.elasticmapreduce.model.PlacementType;
import com.amazonaws.services.elasticmapreduce.model.RunJobFlowRequest;
import com.amazonaws.services.elasticmapreduce.model.RunJobFlowResult;
import com.amazonaws.services.elasticmapreduce.model.StepConfig;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.DeleteObjectsRequest;
import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;

public class EMRUtil {
	private static Properties mprops = null;

	private static Properties getProperties() throws IOException {
		if (mprops != null) {
			return mprops;
		}
		InputStream in = HadoopUtil.class.getClassLoader().getResourceAsStream(
				"hadoop-config.properties");
		mprops = new Properties();
		mprops.load(in);
		return mprops;
	}

	private static PropertiesCredentials mCredentials = null;

	private static PropertiesCredentials getCredentials() throws IOException {
		if (mCredentials != null) {
			return mCredentials;
		}
		mCredentials = new PropertiesCredentials(EMRUtil.class.getClassLoader()
				.getResourceAsStream("AwsCredentials.properties"));
		return mCredentials;
	}

	public static void writeInputFile(String folderName, Goban goban,
			GoState goType, List<GobanNode> rootList, int level)
			throws IOException {
		AmazonS3 s3 = new AmazonS3Client(getCredentials());
		JobConf conf = new JobConf();
		Properties props = getProperties();
		String bucketName = props.getProperty("bucketname");
		for (GobanNode rootNode : rootList) {
			GobanParam param = new GobanParam();
			param.setGoban(goban);
			param.setGoType(goType);
			param.setLevel(level);
			param.setX(rootNode.getX());
			param.setY(rootNode.getY());
			Text key = new Text(rootNode.getX() + "_" + rootNode.getY());
			File file = File.createTempFile(folderName+"_input_" + key, ".txt");
			FSDataOutputStream out = new FSDataOutputStream(
					new FileOutputStream(file), new Statistics(""));
			Writer fwriter = SequenceFile.createWriter(conf, out, Text.class,
					GobanParam.class, CompressionType.NONE, new DefaultCodec());
			fwriter.append(key, param);
			fwriter.close();
			s3.putObject(bucketName, folderName + "/input/input_" + key, file);
			file.delete();
		}
	}

	public static Map<String, String> run(String folderName, Goban goban,
			GoState goType, List<GobanNode> rootList, int level)
			throws IOException, InterruptedException {
		Properties props = getProperties();
		String bucketName = props.getProperty("bucketname");

		writeInputFile(folderName, goban, goType, rootList, level);

		JobFlowInstancesConfig instances = new JobFlowInstancesConfig();
		instances.setHadoopVersion("0.20");
		instances.setInstanceCount(Integer.parseInt(props
				.getProperty("s3.instance.number")));
		instances.setMasterInstanceType(InstanceType.M1Small.toString());
		instances.setSlaveInstanceType(InstanceType.M1Small.toString());
		PlacementType placement = new PlacementType();
		placement.setAvailabilityZone("ap-northeast-1a");
		instances.setPlacement(placement);

		HadoopJarStepConfig jarConfig = new HadoopJarStepConfig(
				props.getProperty("s3.jarpath"));
		jarConfig.setMainClass("jp.co.worksap.ate.gomoku.hadoopai.EMRMain");
		jarConfig.setArgs(Arrays.asList(new String[] {
				"s3://" + bucketName + "/" + folderName + "/input",
				"s3://" + bucketName + "/" + folderName + "/output" }));

		StepConfig stepConfig = new StepConfig("gomoku-" + folderName,
				jarConfig);

		RunJobFlowRequest request = new RunJobFlowRequest("gomoku-"
				+ folderName, instances);
		request.setLogUri("s3://" + bucketName + "/" + folderName + "/logs");
		request.setSteps(Arrays.asList(new StepConfig[] { stepConfig }));

		AmazonElasticMapReduce emr = new AmazonElasticMapReduceClient(
				getCredentials());
		emr.setEndpoint("elasticmapreduce.ap-northeast-1.amazonaws.com");
		RunJobFlowResult result = emr.runJobFlow(request);
		waitForFinish(emr, result);
		Map<String, String> map = readOutputFile(folderName);
		deleteGomokuFolder(folderName);
		return map;
	}

	private static void waitForFinish(AmazonElasticMapReduce emr,
			RunJobFlowResult result) throws InterruptedException {
		JobFlowExecutionState lastState = null;
		STATUS_LOOP: while (true) {
			DescribeJobFlowsRequest desc = new DescribeJobFlowsRequest(
					Arrays.asList(new String[] { result.getJobFlowId() }));
			try {
				DescribeJobFlowsResult descResult = emr.describeJobFlows(desc);
				for (JobFlowDetail detail : descResult.getJobFlows()) {
					String state = detail.getExecutionStatusDetail().getState();
					JobFlowExecutionState jobState = JobFlowExecutionState
							.fromValue(state);
					if (isDone(state)) {
						System.out.println("Job " + state + ": "
								+ detail.toString());
						if (jobState.equals(JobFlowExecutionState.COMPLETED)) {
							break STATUS_LOOP;
						} else {
							throw new RuntimeException("EMR Task Failed!");
						}
					} else if (!jobState.equals(lastState)) {
						lastState = jobState;
						System.out.println("Job " + jobState.name() + " at "
								+ new Date().toString());
					}
				}
			} catch (AmazonServiceException e) {
				e.printStackTrace();
			} catch (AmazonClientException e) {
				e.printStackTrace();
			}
			Thread.sleep(5000);
		}
	}

	private static final List<JobFlowExecutionState> DONE_STATES = Arrays
			.asList(new JobFlowExecutionState[] {
					JobFlowExecutionState.COMPLETED,
					JobFlowExecutionState.FAILED,
					JobFlowExecutionState.TERMINATED });

	private static boolean isDone(String value) {
		JobFlowExecutionState state = JobFlowExecutionState.fromValue(value);
		return DONE_STATES.contains(state);
	}

	public static void deleteGomokuFolder(String folderName) throws IOException {
		AmazonS3 s3 = new AmazonS3Client(getCredentials());
		Properties props = getProperties();
		String bucketName = props.getProperty("bucketname");
		DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(
				bucketName);
		ObjectListing objectList = s3.listObjects(bucketName, folderName);
		List<S3ObjectSummary> summaryList = objectList.getObjectSummaries();
		List<KeyVersion> keyVersionList = new ArrayList<KeyVersion>();
		for (S3ObjectSummary summary : summaryList) {
			KeyVersion keyVersion = new KeyVersion(summary.getKey());
			keyVersionList.add(keyVersion);
		}
		deleteRequest.setKeys(keyVersionList);
		s3.deleteObjects(deleteRequest);
	}

	public static Map<String, String> readOutputFile(String folderName)
			throws IOException {
		Map<String, String> map = new HashMap<String, String>();
		AmazonS3 s3 = new AmazonS3Client(getCredentials());
		Properties props = getProperties();
		String bucketName = props.getProperty("bucketname");
		ObjectListing objectList = s3.listObjects(bucketName, folderName
				+ "/output");
		List<S3ObjectSummary> summaryList = objectList.getObjectSummaries();
		for (S3ObjectSummary summary : summaryList) {
			String key = summary.getKey();
			S3Object object = s3
					.getObject(new GetObjectRequest(bucketName, key));
			BufferedReader freader = new BufferedReader(new InputStreamReader(
					object.getObjectContent()));
			String line;
			while ((line = freader.readLine()) != null) {
				String[] parts = line.split("\\s+");
				map.put(parts[0], parts[1]);
			}
			freader.close();
		}
		return map;
	}

	public static void main(String[] args) throws IOException,
			InterruptedException {
		Goban goban = new Goban(19, 19);
		goban.putBlack(10, 10);
		goban.putBlack(10, 11);
		goban.putWhite(9, 10);
		List<GoNode> nodeList = LocalAIUtil
				.getAllValueableEmptyGoNode(goban, 2);
		List<GobanNode> rootList = new ArrayList<GobanNode>();
		GoState goType = GoState.WHITE;
		for (GoNode goNode : nodeList) {
			GobanNode rootNode = new GobanNode(goban.clone(), goType,
					goNode.getX(), goNode.getY());
			rootList.add(rootNode);
		}
		Map<String, String> map = run("test", goban, goType, rootList, 3);
		for (GobanNode goNode : rootList) {
			String key = goNode.getX() + "_" + goNode.getY();
			String score = map.get(key);
			System.out.println(goNode.getX() + " " + goNode.getY() + ":"
					+ score);
		}
	}
}
